package rpc

import (
	"cluster-cache/store"
	"fmt"
	"github.com/panjf2000/gnet/v2"
	"github.com/panjf2000/gnet/v2/pkg/logging"
	"sync/atomic"
)

type TcpServerV2 struct {
	*gnet.BuiltinEventEngine
	eng          gnet.Engine
	network      string
	addr         string
	connected    int32
	disconnected int32
	handler      Handler
}

func NewTcpServerV2(addr string, hand Handler) *TcpServerV2 {
	return &TcpServerV2{
		network:      "tcp",
		addr:         addr,
		connected:    0,
		disconnected: 0,
		handler:      hand,
	}
}

func (s *TcpServerV2) OnBoot(eng gnet.Engine) (action gnet.Action) {
	logging.Infof("running server on %s", fmt.Sprintf("%s://%s", s.network, s.addr))
	s.eng = eng
	return
}

func (s *TcpServerV2) OnOpen(c gnet.Conn) (out []byte, action gnet.Action) {
	code := HeadBodyCodecPool.Get().(*HeadBodyCodec)
	code.Reset()
	c.SetContext(code)
	atomic.AddInt32(&s.connected, 1)
	return
}

func (s *TcpServerV2) OnClose(c gnet.Conn, err error) (action gnet.Action) {
	if err != nil {
		logging.Infof("error occurred on connection=%s, %v\n", c.RemoteAddr().String(), err)
	}
	disconnected := atomic.AddInt32(&s.disconnected, 1)
	connected := atomic.AddInt32(&s.connected, -1)
	if connected == 0 {
		logging.Infof("all %d connections are closed", disconnected)
	}
	HeadBodyCodecPool.Put(c.Context())
	return
}

func (s *TcpServerV2) OnTraffic(c gnet.Conn) (action gnet.Action) {
	code := c.Context().(*HeadBodyCodec)
	var packets [][]byte
	futures := make([]CallFuture, 0)
	for {
		data, seq, err := code.Decode(c)
		if err != nil {
			if err == ErrIncompletePacket {
				break
			}
			code.Reset()
			return
		}
		future := s.handler.Accept(data)
		futures = append(futures, future...)
		packet, _ := EncodeWithSeq(store.EmptyByte, seq)
		code.Reset()
		packets = append(packets, packet)
	}
	for i := range futures {
		futures[i].WaitDone()
	}
	if n := len(packets); n > 1 {
		_, _ = c.Writev(packets)
	} else if n == 1 {
		_, _ = c.Write(packets[0])
	}
	return
}

func (s *TcpServerV2) Start() {
	err := gnet.Run(s, s.network+"://"+s.addr,
		gnet.WithMulticore(true),
		gnet.WithReusePort(true),
		gnet.WithSocketRecvBuffer(BufferSize8k),
		gnet.WithSocketSendBuffer(BufferSize8k),
		gnet.WithTCPNoDelay(gnet.TCPNoDelay),
	)
	logging.Infof("server exits with error: %v", err)
}
